package workerpool

import (
	"context"
	"sync"
	"time"

	_err "github.com/romapres2010/goapp/pkg/common/error"
	_log "github.com/romapres2010/goapp/pkg/common/logger"
	_metrics "github.com/romapres2010/goapp/pkg/common/metrics"
	_recover "github.com/romapres2010/goapp/pkg/common/recover"
)

// WorkerState - 枚举定义了worker在其生命周期中可能处于的不同状态：
type WorkerState int

const (
	WORKER_STATE_NEW                           WorkerState = iota // worker 已创建。
	WORKER_STATE_WORKING                                          // worker 正在处理任务。
	WORKER_STATE_IDLE                                             // worker 处于空闲状态。
	WORKER_STATE_TERMINATING_PARENT_CTX_CLOSED                    // 由于父上下文关闭，worker正在停止
	WORKER_STATE_TERMINATING_STOP_SIGNAL                          // 由于接收到停止信号，worker正在停止
	WORKER_STATE_TERMINATING_TASK_CH_CLOSED                       // 由于任务通道关闭，worker正在停止
	WORKER_STATE_TERMINATED                                       // worker已停止
	WORKER_STATE_RECOVER_ERR                                      // worker因错误而停止。
)

// Worker - 结构体
// Worker 结构体表示执行任务的工作器，包含以下字段：
type Worker struct {
	pool *Pool // 所属的池

	parentCtx context.Context //池的父上下文，工作器在此上下文中工作。

	externalId uint64           // 工作器的外部标识符，用于日志记录。
	stopCh     chan interface{} // 来自“外部世界”的停止工作器的命令通道。

	id      uint                // worker编号，用于日志记录。
	state   WorkerState         // worker生命周期的状态
	errCh   chan<- *WorkerError // 用于向Pool报告Worker关键错误的通道。
	timeout time.Duration       // 任务执行的最大等待时间，启动任务时传递给任务。

	taskQueueCh   <-chan *Task //  任务队列的通道。
	taskInProcess *Task        // 当前正在处理的任务。

	mx sync.RWMutex // 读写锁，用于同步对工作器状态的访问。
}

// setState - 设置worker生命周期的状态
func (wr *Worker) setState(state WorkerState) {
	wr.mx.Lock()
	defer wr.mx.Unlock()
	wr.setStateUnsafe(state)
}

// setStateUnsafe - 设置worker生命周期的状态(未通过读写锁实现线程安全)
func (wr *Worker) setStateUnsafe(state WorkerState) {
	wr.state = state
}

// GetState - 获取worker生命周期的状态
func (wr *Worker) GetState() WorkerState {
	wr.mx.RLock()
	defer wr.mx.RUnlock()
	return wr.getStateUnsafe()
}

// getStateUnsafe - 获取worker生命周期的状态(未通过读写锁实现线程安全)
func (wr *Worker) getStateUnsafe() WorkerState {
	return wr.state
}

// newWorker - 创建一个新的Worker实例
func newWorker(parentCtx context.Context, pool *Pool, taskQueueCh <-chan *Task, id uint, externalId uint64, errCh chan<- *WorkerError, timeout time.Duration) *Worker {
	var worker = Worker{
		parentCtx:   parentCtx,
		id:          id,
		pool:        pool,
		externalId:  externalId,
		taskQueueCh: taskQueueCh,
		timeout:     timeout,
		errCh:       errCh,
	}

	worker.setState(WORKER_STATE_NEW)

	return &worker
}

// run - 启动Worker
func (wr *Worker) run(wg *sync.WaitGroup) {

	// 在启动期间锁定工作器，以避免同时使用同一个指针
	if wr.mx.TryLock() { // 不推荐使用 TryLock，但在这种情况下它非常方便
		defer wr.mx.Unlock()
	} else {
		//_log.Info("Worker - already locked for run: PoolName, WorkerId, WorkerExternalId, State", wr.pool.name, wr.id, wr.externalId, wr.state)
		err := _err.NewTyped(_err.ERR_WORKER_POOL_ALREADY_LOCKED, wr.externalId, wr.id, wr.state).PrintfError()

		wr.errCh <- &WorkerError{ //将err发送到池的公共错误通道
			err:    err,
			worker: wr,
		}
		return
	}

	// 只能启动新的工作器或者在发生恐慌（panic）之后启动
	if wr.state == WORKER_STATE_NEW || wr.state == WORKER_STATE_RECOVER_ERR {
		wr.setStateUnsafe(WORKER_STATE_IDLE) // 将工作器设置为空闲状态
		//_log.Debug("Worker - START: PoolName, WorkerId, WorkerExternalId, State", wr.pool.name, wr.id, wr.externalId, wr.state)
	} else {
		//_log.Info("Worker - has incorrect state to run: PoolName, WorkerId, WorkerExternalId, State", wr.pool.name, wr.id, wr.externalId, wr.state)
		err := _err.NewTyped(_err.ERR_WORKER_POOL_RUN_INCORRECT_STATE, wr.externalId, wr.id, wr.state, "NEW', 'RECOVER_ERR").PrintfError()

		wr.errCh <- &WorkerError{ // 将err发送到池的公共错误通道
			err:    err,
			worker: wr,
		}
		return
	}

	// 我们创建一个内部通道，用于通知工作器需要从“外部世界”紧急停止
	// 在编程中，特别是在涉及并发和多线程的环境下，创建这样的内部通道（或信号机制）是为了使外部组件能够对工作器（worker）发出停止当前任务的请求。这在需要优雅地关闭工作器或处理取消操作时非常有用
	wr.stopCh = make(chan interface{}, 1)
	//defer close(wr.stopCh) 我们将在发送信号的地方关闭通道

	// 处理worker的panic
	defer func() {
		if r := recover(); r != nil {
			//_log.Info("Worker - RECOVER FROM PANIC: PoolName, WorkerId, WorkerExternalId, State", wr.pool.name, wr.id, wr.externalId, wr.state)
			err := _recover.GetRecoverError(r, wr.externalId)
			if err != nil {
				wr.setStateUnsafe(WORKER_STATE_RECOVER_ERR)

				wr.errCh <- &WorkerError{ // 将err发送到池的公共错误通道
					err:    err,
					worker: wr,
				}
			}
		} else {
			wr.setStateUnsafe(WORKER_STATE_TERMINATED)
		}

		// 如果在 WaitGroup 的范围内工作，那么将减少计数器
		// 在 Go 语言中，sync.WaitGroup（简称 WaitGroup）是一个用于同步并发操作的同步原语。它的工作原理是使用一个计数器来跟踪一组并发操作的完成情况。
		// 当一个 goroutine 开始一个并发操作时，它会调用 WaitGroup 的 Add(1) 方法来增加计数器。
		// 当操作完成时，它会调用 Done() 方法来减少计数器。当 WaitGroup 的计数器达到零时，任何调用 Wait() 方法的 goroutine 将会继续执行，表示它正在等待的所有并发操作都已完成
		if wg != nil {
			wg.Done()
		}
	}()

	//从任务队列通道 taskQueueCh 等待任务（忽略空任务），停止信号或池的父上下文 pool 的关闭
	//在 Go 语言中，这通常涉及到使用 select 语句来同时监听多个通道。当其中一个通道准备就绪（即有数据可读取）时，select 语句就会执行相应的分支代码
	for {
		//对于 select 语句，没有保证说通道会按照它们被写出的顺序进行查询
		//因此，在每次新的迭代中，我们首先检查工作器是否未被停止
		select {
		case _, ok := <-wr.stopCh:
			if ok { // 通道已打开，我们收到了停止命令
				//_log.Info("Worker - STOP - got quit signal: PoolName, WorkerId, WorkerExternalId", wr.pool.name, wr.id, wr.externalId)
				wr.setStateUnsafe(WORKER_STATE_TERMINATING_STOP_SIGNAL)
			} else {
				// 内部逻辑出现了不恰当的情况 - 我们记录日志以供分析
				_log.Error("Worker - STOP - stop chanel closed: PoolName, WorkerId, WorkerExternalId", wr.pool.name, wr.id, wr.externalId)
			}
			return
		case <-wr.parentCtx.Done():
			// 父上下文已关闭
			//_log.Info("Worker - STOP - got parent context close: PoolName, WorkerId, WorkerExternalId", wr.pool.name, wr.id, wr.externalId)
			wr.setStateUnsafe(WORKER_STATE_TERMINATING_PARENT_CTX_CLOSED)
			return
		default:
		}

		// Если worker не остановлен, то проверяем канал-очереди задач
		select {
		case task, ok := <-wr.taskQueueCh:
			if ok { // канал очереди задач открыт
				if task != nil { // игнорируем пустые задачи
					//_log.Debug("Worker - got task: PoolName, WorkerId, WorkerExternalId, TaskName", wr.pool.name, wr.id, wr.externalId, task.name)
					_metrics.IncWPWorkerProcessCountVec(wr.pool.name)                               // Метрика - количество worker в работе
					_metrics.SetWPTaskQueueBufferLenVec(wr.pool.name, float64(len(wr.taskQueueCh))) // Метрика - длина необработанной очереди задач

					wr.setStateUnsafe(WORKER_STATE_WORKING)
					wr.taskInProcess = task

					//_log.Debug("Worker - start to process task: PoolName, WorkerId, WorkerExternalId, TaskName", wr.pool.name, wr.id, wr.externalId, task.name)
					task.process(wr.id, wr.timeout)

					wr.taskInProcess = nil
					wr.setStateUnsafe(WORKER_STATE_IDLE)

					_metrics.DecWPWorkerProcessCountVec(wr.pool.name)                            // Метрика - количество worker в работе
					_metrics.IncWPTaskProcessDurationVec(wr.pool.name, task.name, task.duration) // Метрика - время выполнения задачи по имени
					//_log.Debug("Worker - end process task: PoolName, WorkerId, WorkerExternalId, TaskName, Duration", wr.pool.name, wr.id, wr.externalId, task.name, task.duration)
				}
			} else { // Если канал-очереди task закрыт - прерываем работу
				//_log.Debug("Worker - task channel was closed: PoolName, WorkerId, WorkerExternalId", wr.pool.name, wr.id, wr.externalId)
				wr.setStateUnsafe(WORKER_STATE_TERMINATING_TASK_CH_CLOSED)
				return
			}
		case _, ok := <-wr.stopCh:
			if ok { // канал был открыт и получили команду на остановку
				//_log.Info("Worker - STOP - got quit signal: PoolName, WorkerId, WorkerExternalId", wr.pool.name, wr.id, wr.externalId)
				wr.setStateUnsafe(WORKER_STATE_TERMINATING_STOP_SIGNAL)
			} else {
				// Не корректная ситуация с внутренней логикой - логируем для анализа
				_log.Error("Worker - STOP - stop chanel closed: PoolName, WorkerId, WorkerExternalId", wr.pool.name, wr.id, wr.externalId)
			}
			return
			//case <-wr.parentCtx.Done():
			//	// закрыт родительский контекст
			//	//_log.Info("Worker - STOP - got parent context close: PoolName, WorkerId, WorkerExternalId", wr.pool.name, wr.id, wr.externalId)
			//	wr.setStateUnsafe(WORKER_STATE_TERMINATING_PARENT_CTX_CLOSED)
			//	return
		}
	}
}

// Stop - принудительная остановка worker, не дожидаясь отработки всей очереди
func (wr *Worker) Stop(shutdownMode PoolShutdownMode) {
	if wr == nil {
		return
	}

	//_log.Debug("Worker - STOP: PoolName, WorkerId, WorkerExternalId, State", wr.pool.name, wr.id, wr.externalId, wr.state)

	// Останавливать можно только в определенных статусах
	if wr.state == WORKER_STATE_NEW || wr.state == WORKER_STATE_WORKING || wr.state == WORKER_STATE_IDLE {
		//_log.Debug("Worker  - send STOP signal: PoolName, WorkerId", wr.pool.name, wr.id)

		// Отправляем сигнал и закрываем канал - если worker ни разу не запускался, то wr.stopCh будет nil
		if wr.stopCh != nil {
			wr.stopCh <- true
			close(wr.stopCh)
		}

		// В режиме "Hard" остановки запускаем прерывание текущей task
		if shutdownMode == POOL_SHUTDOWN_HARD {
			if wr.taskInProcess != nil {
				wr.taskInProcess.Stop()
			}
		}
	}
}
